package org.jpush.core;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.jpush.impl.xml.XMLPacket;
import org.jpush.packages.AnalyticsProtocol;
import org.jpush.packages.IncomingPacket;
import org.jpush.packages.OutgoingPacket;
import org.jpush.utils.DataBuffer;
import org.jpush.utils.Pointer;
import org.jpush.utils.StopWatch;
import org.jpush.utils.Utilities;

public class Dispatcher
{
	protected Server server;

	private ConcurrentMap<Integer, Service> serviceMap = 
			new ConcurrentHashMap<Integer, Service>();

	private ConcurrentMap<Long, Service> workerServiceMap = 
			new ConcurrentHashMap<Long, Service>();
	
	// Store socket and physical connection
//	private ConcurrentMap<JPushClientSocket, PhysicalConnection> connectionMap 
//			= new ConcurrentHashMap<JPushClientSocket, PhysicalConnection>();
	
	private static int observerId;
	
	public Dispatcher(Server server)
	{
		this.server = server;
	}

	public void onWriteComplete(PhysicalConnection channel, int dwIoSize)
	{
	    if (!channel.isObserverChannel())
	    {
	    	server.getServerStats().addToCumul(ServerStats.Measures.BandwidthOutbound, dwIoSize);
	    	// TODO hashcode is OK?
	    	server.getServerStats().addToKeyedDuration(ServerStats.Measures.BandwidthOutboundPerConnection, /*(int)*/ channel.hashCode(), dwIoSize);
	    }

	    Pointer<Boolean> pBlnIsBufferIdle = new Pointer<Boolean>(Boolean.FALSE);

	    PhysicalConnection.ConnectionStatus status = channel.onSendCompleted(dwIoSize, pBlnIsBufferIdle);

	    if (status == PhysicalConnection.ConnectionStatus.Attached && pBlnIsBufferIdle.getValue().booleanValue())
	    {
	        if (channel.isObserverChannel())
	        {
	        	server.getMonitorsBroadcastManager().HandleSingleMonitor(channel);
	        }
	    }

	    channel.decrementIoWorkersReferenceCounter();
	}

	/**
	 * This will be called when the PhysicalConnection receives data
	 * @param connection
	 * @param dwIoSize
	 */
	public void onReceiveComplete(PhysicalConnection connection, int dwIoSize)
	{
	    boolean isClient = !connection.isObserverChannel();

	    if (isClient)
	    {
	    	server.getServerStats().addToCumul(ServerStats.Measures.BandwidthInbound, dwIoSize);
	    	server.getServerStats().addToKeyedDuration(ServerStats.Measures.BandwidthInboundPerConnection, /*(unsigned int)*/ connection.hashCode(), dwIoSize);
	    }

	    /* Read the data into the buffer from system buffer */
	    PhysicalConnection.ConnectionStatus status = connection.readReceivedBytes();

	    // If the state of the connection is normal for state of receiving data
	    // the worker thread will not process it so decrement the counter of
	    // worker thread for this connection
	    if(status == PhysicalConnection.ConnectionStatus.Disposable || 
	    		status == PhysicalConnection.ConnectionStatus.WaitingForWrite)
	    {
	        connection.decrementIoWorkersReferenceCounter();
	        return;
	    }

	    // The connection is either connected or attached.
	    if (dwIoSize == 0) //Peer wants to close the connection.(No data is read)
	    {
	        connection.close(false);
	        if (connection.getStatus() == PhysicalConnection.ConnectionStatus.Attached)
	        {
	            if (isClient)
	            {
	            	server.getConnectionEventListener().onConnectionDisconnected(connection.getLogicalConnection());
	            }
	            else
	            {
	                //TODO notify observer is out.
	            }
	        }
	        connection.decrementIoWorkersReferenceCounter();
	        return;
	    }


	    //The Processing Loop.
	    //int uCommandID;
	    Pointer<Integer> pCommandID = new Pointer<Integer>();
	    Pointer<IncomingPacket> pInComingPacket = new Pointer<IncomingPacket>();
	    Pointer<Integer> pExtractedBytes = new Pointer<Integer>();
	    Protocol.Result iResult = Protocol.Result.Success;
	    //int uExtractedBytes;
	    Protocol pProtocol = connection.getProtocol();
	    DataBuffer sourceBuffer = connection.getReceiveBuffer();

//	    boolean blnProcessDataInQueue = true;
//	    
//	    // Loop to read 
//	    while (blnProcessDataInQueue)
//	    {
//	        //watch.GetElapsedTime(false);
	    	// Try to deserialize data into packet
	        iResult = pProtocol.tryDeserializeIncomingPacket(sourceBuffer, pInComingPacket, 
	        		pCommandID, pExtractedBytes, connection.getConnectionContext());
	        
	        int uCommandID = 0;
	        if (pCommandID.notNull())
	        	uCommandID = pCommandID.getValue().intValue();
	
	        int uExtractedBytes = 0;
	        if (pExtractedBytes.notNull())
	        	uExtractedBytes = pExtractedBytes.getValue().intValue();
	        
	        IncomingPacket packet = pInComingPacket.getValue();
	        if (iResult == Protocol.Result.Success)
	        {
	        	// Clear the data from buffer because it is read and saved into the packet
	            connection.getReceiveBuffer().pop(uExtractedBytes);
	            
	            // Check the status of connection
	            status  = connection.getStatus();
	            if (status == PhysicalConnection.ConnectionStatus.Attached)
	            {
	                if(isClient)
	                    dispatchRequest(uCommandID, connection.getLogicalConnection(), packet, uExtractedBytes);
	                else
	                    handleMonitorRequest(connection, packet);
	            }
	            else if(status == PhysicalConnection.ConnectionStatus.Connected)
	            {
	                if(isClient)
	                    processFirstPacket(connection, uCommandID, packet, uExtractedBytes);
	                else
	                    processMonitorFirstPacket(connection, packet);
	            }
//	            else
//	            {
//	                //Status changed by another thread eg .disconnect.
//	                blnProcessDataInQueue = false;
//	            }
	            pProtocol.disposeIncomingPacket(packet);
	        }
	        else if (iResult == Protocol.Result.eDecodingFailure)
	        {
	            connection.getReceiveBuffer().pop(uExtractedBytes);
	        }
//	        else
//	        {
//	            break;
//	        }
//	    }
	    //
	    if (!PhysicalConnection.notConnected(connection.getStatus()) &&
	    		iResult == Protocol.Result.eIncompletePacket)
	    {
	        connection.postReceive();
	    }

	    connection.decrementIoWorkersReferenceCounter();
	}

//	/**
//	 * Register the socket and store the connection
//	 * 
//	 * @param socket
//	 * @param physicalConnection
//	 * @return true if succeeded
//	 */
//	public boolean addSocketContext(JPushClientSocket socket, PhysicalConnection physicalConnection)
//	{
//		// Initialization
//		onInitializeReady(physicalConnection);
//		
//		// Register
//		connectionMap.put(socket, physicalConnection);
//		
//		return true;
//	}

	/**
	 * This is call when the first time the client socket is created.
	 * @param connection
	 */
	public void onInitializeReady(PhysicalConnection connection)
	{
	    boolean isClient = !connection.isObserverChannel();

	    if (isClient)
	    {
	        Pointer<ConnectionContext> pContext = new Pointer<ConnectionContext>();
	        OutgoingPacket packet = null;
	        
	        // Get the connection context and the response packet
	        packet = server.getConnectionEventListener().onNewConnection(pContext);
	        
	        // TODO, what is the condition?
	        if (pContext.notNull())
	            connection.setConnectionContext(pContext.getValue());

	        if (packet != null)
	        {
	            connection.pushPacket(packet);
	            server.disposeOutgoingPacket(packet);
	        }
	    }

	    connection.postReceive();
	}

	public void registerService(int uCommmand, Service service, String serviceName)
	{
	    //serviceMap[uCommmand] = pServiceInfo;
		service.setName(serviceName);
	    serviceMap.put(new Integer(uCommmand), service);
	}

	public String getServiceNames()
	{
		StringBuffer sb = new StringBuffer();
		for (Service si : serviceMap.values())
		{
			sb.append("<service val=\"").append(si.getName()).append("\"/>");
		}
		
		return sb.toString();
	}

	public void setCurrentService(Service service)
	{
	    workerServiceMap.put(new Long(Thread.currentThread().getId()), service);
	}

	public void UnsetCurrentService()
	{
	    workerServiceMap.remove(new Long(Thread.currentThread().getId()));
	}

	public Service getCurrentService()
	{
    	Long lThreadId = new Long(Thread.currentThread().getId());
	    return workerServiceMap.get(lThreadId);
	}

	public void NotifyObserversClientIN(String key, String peerIP, int peerPort)
	{
//	    String timestamp = Utilities.getCurrentTime();
//
//	    XMLPacket *response = new XMLPacket(AnalyticsProtocol.VisitorInResponse);
//	    response.setArgumentAsText("time", timestamp.c_str());
//	    response.setArgumentAsText("name", key);
//	    response.setArgumentAsText("ip", peerIP.c_str());
//	    response.setArgumentAsInt("port", peerPort);
		String timestamp = Utilities.getCurrentTime();
	    XMLPacket response = new XMLPacket(AnalyticsProtocol.VisitorInResponse);
	    response.setArgumentAsText("time", timestamp);
	    response.setArgumentAsText("name", key);
	    response.setArgumentAsText("ip", peerIP);
	    response.setArgumentAsInt("port", peerPort);

//	    monitorBroadcastManager.PushPacket(response, "clientsIn", key, 0);
	    server.getMonitorsBroadcastManager().pushPacket(response, "clientsIn", key, 0);
	}

	public void NotifyObserversClientOut(String key)
	{
	    XMLPacket response = new XMLPacket(AnalyticsProtocol.VisitorOutResponse);
	    response.setArgumentAsText("name", key);

//	    monitorBroadcastManager.PushPacket(response, "clientsOut");
	    server.getMonitorsBroadcastManager().pushPacket(response, "clientsOut");

	    //Remove client from the other broadcast group :
//	    monitorBroadcastManager.RemovePacket(key, 0, "clientsIn");
	    server.getMonitorsBroadcastManager().removePacket(key, 0, "clientsIn");
	}
//	private:
	private void dispatchRequest(int uCommand, LogicalConnection pClient,IncomingPacket packet,int serviceBytes)
	{
	    //StopWatch dispatchWatch(m_QPFrequency);

//	    serviceMapT.iterator it = serviceMap.find(uCommand);
//	    if (it == serviceMap.end())
//	        return;
		Service service = serviceMap.get(new Integer(uCommand));
		if (service == null)
			return;
	    //
	    //Service* pHandler = it.second.pService;
		//Service pHandler = serviceInfo.service;

	    //wcout << L"Locating Service : " << dispatchWatch.GetElapsedTime(false) << std.endl;

	    //Mark dispatched service :

	    //setCurrentService(it.second.serviceName);
		setCurrentService(service);

	    StopWatch watch = new StopWatch();
	    service.handle(pClient, packet);


	    double duration = watch.getElapsedTime();
	    /*	wcout << L"Service Time : " << watch.GetElapsedTime() << std.endl;
	     */


	    //StopWatch statsClock(m_QPFrequency);
//	    stats.addToDistribution(ServerStats.PerformanceProcessingTimePerService, serviceInfo.serviceName, duration);
	    server.getServerStats().addToDistribution(ServerStats.Measures.PerformanceProcessingTimePerService, 
	    		service.getName(), duration);
	    //wcout << L"Stat 1 : " << statsClock.GetElapsedTime(false) << std.endl;

//	    stats.addToDuration(ServerStats.PerformanceProcessingTime, duration);
	    server.getServerStats().addToDuration(ServerStats.Measures.PerformanceProcessingTime, duration);
	    //wcout << L"Stat 2 : " << statsClock.GetElapsedTime(false) << std.endl;

	    UnsetCurrentService();

	    //Stats. :

//	    stats.addToDistribution(ServerStats.BandwidthInboundVolPerRequest, serviceInfo.serviceName, serviceBytes);
	    server.getServerStats().addToDistribution(ServerStats.Measures.BandwidthInboundVolPerRequest, 
	    		service.getName(), serviceBytes);
	    //wcout << L"Stat 3 : " << statsClock.GetElapsedTime(false) << std.endl;


//	    stats.addToDistribution(ServerStats.PerformanceRequestVolPerRequest, serviceInfo.serviceName, 1);
	    server.getServerStats().addToDistribution(ServerStats.Measures.PerformanceRequestVolPerRequest, 
	    		service.getName(), 1);
	    //wcout << L"Stat 4 : " << statsClock.GetElapsedTime(false) << std.endl;


	    //wcout << L"Dispatch Time : " << dispatchWatch.GetElapsedTime() << std.endl;
	}

	private void processFirstPacket(PhysicalConnection connection,int uCommand, IncomingPacket packet,int serviceBytes)
	{
	    Pointer<OutgoingPacket> pOutPacket = new Pointer<OutgoingPacket>();
	    //
	    Pointer<LogicalConnection> pClient = new Pointer<LogicalConnection>();
	    ConnectionEventListener.Result iResult = server.getConnectionEventListener().onFirstRequest(packet, connection.getConnectionContext(), pClient, pOutPacket);

	    LogicalConnection client = pClient.getValue();
	    //packet and lpContext are not good.
	    if (iResult == ConnectionEventListener.Result.RefuseAndClose)
	    {
	        if (pOutPacket.notNull())
	        {
	            connection.pushPacket(pOutPacket.getValue());
	            server.disposeOutgoingPacket(pOutPacket.getValue());
	        }
	        connection.close(false);
	        return;
	    }
	    //
	    if (iResult == ConnectionEventListener.Result.RefuseRequest)
	    {
	        if (pOutPacket.notNull())
	        {
	            connection.pushPacket(pOutPacket.getValue());
	            server.disposeOutgoingPacket(pOutPacket.getValue());
	        }
	        return;
	    }

	    String clientKey = client.getKey();


	    //iResult >= CClientFactory.CreateClient
//	    if (!channelFactory.RemoveFromUnattachedList(pChannel))
	    if (!server.getChannelFactory().removeFromUnattachedList(connection))
	    {
//	        delete pClient;
	        connection.decrementIoWorkersReferenceCounter();
	        return;
	    }


	    //Bind the two structures.
	    client.physicalConnection = connection;
	    connection.attachToClient(client); //status is attached.

//	    stats.addToCumul(ServerStats.VisitorsHitsIn, 1);
	    server.getServerStats().addToCumul(ServerStats.Measures.VisitorsHitsIn, 1);

//	    LogicalConnection pRet = clientFactoryImpl.AddLogicalConnection(pClient);
	    LogicalConnection pRet = server.getClientFactory().addLogicalConnection(client);
	    if (pRet != client)
	    {
	        //Copy data from existing to new instance.
	    	server.getConnectionEventListener().copyLogicalConnection(pRet, client);

	        //Silent disconnect :
//	        clientFactoryImpl.SilentDisconnect(pRet);
	    	server.getClientFactory().silentDisconnect(pRet);

	        //Now add ours :
//	        clientFactoryImpl.AddLogicalConnection(pClient);
	    	server.getClientFactory().addLogicalConnection(client);

	        //Fire Reconnect event.
	        server.getConnectionEventListener().onConnectionReconnected(client);

	        //Add client to a streamer :
//	        broadcastStreamerManager.AddClient(pClient);
	        server.getBroadcastStreamerManager().addConnection(client);

	        client.decrementUsage();
	        return;
	    }

	    //Fire connect event.
	    server.getConnectionEventListener().onConnectionConnected(client);
	    //Add client to a streamer :
//	    broadcastStreamerManager.AddClient(pClient);
	    server.getBroadcastStreamerManager().addConnection(client);

	    //Statistics :
	    NotifyObserversClientIN(clientKey, connection.getPeerIP(), connection.getPeerPort());
//	    stats.addToCumul(ServerStats.VisitorsOnline, 1);
	    server.getServerStats().addToCumul(ServerStats.Measures.VisitorsOnline, 1);


	    if (iResult == ConnectionEventListener.Result.CreateAndRouteRequest)
	    {
	        dispatchRequest(uCommand, client, packet, serviceBytes);
	    }
	    else
	    {
	        if (pOutPacket.notNull())
	        {
	            connection.pushPacket(pOutPacket.getValue());
	            server.disposeOutgoingPacket(pOutPacket.getValue());
	        }
	    }

	    client.decrementUsage();
	}

	private void handleMonitorRequest(PhysicalConnection pChannel, IncomingPacket packet)
	{
	    //XMLPacket& requestPacket = (XMLPacket&) packet;
		if (!(packet instanceof XMLPacket))
		{
			return;
		}
		
		XMLPacket requestPacket = (XMLPacket) packet;

		AnalyticsProtocol typeId = requestPacket.getTypeId();

	    if (typeId == AnalyticsProtocol.LiveSubscriptionRequest)
	    {
	        boolean bSubscribe = requestPacket.getArgumentAsBool("resume");
	        if (bSubscribe)
	        {
	            OutgoingPacket pInitPacket = server.getServerStats().getInitializationPacket();
	            pChannel.pushPacket(pInitPacket);
	            server.disposeOutgoingPacket(pInitPacket);

	            //
//	            monitorBroadcastManager.SubscribeClient(pChannel.getLogicalConnectionKey(), "stats");
//	            monitorBroadcastManager.SubscribeClient(pChannel.getLogicalConnectionKey(), "clientsIn");
//	            monitorBroadcastManager.SubscribeClient(pChannel.getLogicalConnectionKey(), "clientsOut");
	            server.getMonitorsBroadcastManager().subscribeConnection(pChannel.getLogicalConnectionKey(), "stats");
	            server.getMonitorsBroadcastManager().subscribeConnection(pChannel.getLogicalConnectionKey(), "clientsIn");
	            server.getMonitorsBroadcastManager().subscribeConnection(pChannel.getLogicalConnectionKey(), "clientsOut");
	        }
	        else
	        {
//	            monitorBroadcastManager.RemoveClient(pChannel.getLogicalConnectionKey());
	        	server.getMonitorsBroadcastManager().removeConnection(pChannel.getLogicalConnectionKey());
	        }
	    }


	    if (typeId == AnalyticsProtocol.LogoutRequest)
	    {
	        pChannel.close(false);
	    }

	    if (typeId == AnalyticsProtocol.ConsoleCommandRequest)
	    {
	        String command = requestPacket.getArgumentAsText("command");

	        XMLPacket response = new XMLPacket(AnalyticsProtocol.ConsoleCommandResponse);

	        response.setArgumentAsText("client", requestPacket.getArgumentAsText("client"));

	        if ("about".equals(command))
	        {
	            String str = server.getServerInfos() + "\nBased on PushFramework version 1.0";
	            response.setArgumentAsText("console", str);
	        }

	        if ("profiling enable".equals(command))
	        {
	            String str;
	            if (server.getProfilingStatus() == true)
	            {
	                str = "Profiling is already enabled.";
	            }
	            else
	            {
	            	server.enableProfiling(-1);
	                str = "Profiling was enabled.";
	            }
	            response.setArgumentAsText("console", str);
	        }

	        if ("profiling disable".equals(command))
	        {
	            String str;
	            if (!server.getProfilingStatus())
	            {
	                str = "Profiling is already disabled.";
	            }
	            else
	            {
	            	server.disableProfiling();
	                str = "Profiling was disabled.";
	            }
	            response.setArgumentAsText("console", str);
	        }

	        if ("profiling status".equals(command))
	        {
	            response.setArgumentAsText("console", !server.getProfilingStatus() ? "Profiling was found to be disabled." :
	                                       "Profiling was found to be enabled.");
	        }

	        //char[] pOut = new char[256];
	        StringBuffer sbOut = new StringBuffer();

	        boolean bRet = server.handleMonitorRequest(command, sbOut);

	        if (bRet)
	        {
	            response.setArgumentAsText("console", sbOut.toString());
	        }

	        pChannel.pushPacket(response);
	    }

	}

	private void processMonitorFirstPacket(PhysicalConnection pChannel, IncomingPacket packet)
	{
	    //XMLPacket& requestPacket = (XMLPacket&) packet;
		if (!(packet instanceof XMLPacket))
		{
			return;
		}
		
		XMLPacket requestPacket = (XMLPacket) packet;

	    //
	    XMLPacket response = new XMLPacket(AnalyticsProtocol.LoginResponse);

	    String password = requestPacket.getArgumentAsText("password");

	    //if (password == pServerImpl.getMonitorPassword())
	    if (Utilities.objectEquals(password, server.getMonitorPassword()))
	    {

	        observerId = 1;
	        observerId++;
	        StringBuffer sb = new StringBuffer();
	        sb.append(observerId);



//	        if (channelFactory.RemoveFromUnattachedList(pChannel))
	        if (server.getChannelFactory().removeFromUnattachedList(pChannel))
	        {
	            pChannel.attachToClient(sb.toString());
	            //
//	            monitorBroadcastManager.AddMonitor(pChannel);
	            server.getMonitorsBroadcastManager().AddMonitor(pChannel);
	            //
	            response.setArgumentAsText("result", "ok");
	            pChannel.pushPacket(response);
	        }
	        else
	        {
	            //Structure was removed by closeUnlogged.
	        }
	    }
	    else
	    {
	        //Reply
	        response = new XMLPacket(AnalyticsProtocol.LoginResponse);
	        response.setArgumentAsText("result", "pass");
	        pChannel.pushPacket(response);
	        //pChannel.Close(true);
	    }
	}

}
